package rhc.discribute.node.connecter.impl.zookper;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

import rhc.discribute.BaseComponent;
import rhc.discribute.SingleInstanceComponent;
import rhc.discribute.node.connecter.impl.ZKPathLockData;
import rhc.discribute.node.connecter.impl.ZookperConnecter;
import rhc.discribute.node.connecter.impl.ZookperLock;
import rhc.discribute.node.connecter.impl.zookper.watchCallback.LatchWatchCallback;
import rhc.discribute.node.connecter.impl.zookper.watchCallback.LockPathWatchCallback;
import rhc.discribute.node.connecter.schedule.LockPathSchedule;
import rhc.discribute.util.ScheduleUpdateUtil;

/**
 * zookper分布式锁实现
 * 
 * @author rhc
 *
 */
public class ZookperLockImpl extends BaseComponent implements ZookperLock {

	/**
	 * 分布式锁的永久节点，当获取某一个锁时在该节点创建临时节点
	 */
	private static final String LOCK_PERSISTENT_PATH = "/lllllk_persistent";

	/**
	 * 生成锁的时候需要创建临时顺序节点，用该字段作为连接锁path和序号
	 */
	private static final String LOCK_SEPARATOR = "_adfdsa_";

	private ZookperConnecter zkConnecter;

	private ZookperOperate zkOperate;

	public ZookperLockImpl(ZookperConnecter zkConnecter, ZookperOperate zkOperate)
			throws UnsupportedEncodingException, Exception {
		this.zkConnecter = zkConnecter;
		this.zkOperate = zkOperate;
		createLockPersistentPath();
	}

	@Override
	public ZKPathLockData tryAcquire(String lockPath) throws Exception {
		String innerLockPath = this.checkLockPath(lockPath);
		// 获取该路径下已经有的锁路径，如果不为空则说明已有其它锁在获取,直接返回
		List<String> alreadyLockPathList = getLockPathAlreadyLockPath(innerLockPath);
		if (null != alreadyLockPathList && alreadyLockPathList.isEmpty()) {
			return null;
		}
		// 尝试创建一个路径锁
		ZKPathLockData zkPLD = this.createLockPath(innerLockPath);
		String currentEphPath = zkPLD.getEphePath();

		if (this.isAlreadyGetLock(innerLockPath, currentEphPath)) {
			addScheduleDelPath(zkPLD,null,null);
			// 如果获取到锁则返回锁信息
			return zkPLD;
		} else {
			// 否则删除创建的临时节点，返回null
			this.release(zkPLD);
			return null;
		}
	}

	@Override
	public ZKPathLockData tryAcquire(String lockPath,long waitSeconds) throws Exception {
		String innerLockPath = this.checkLockPath(lockPath);
		// 尝试创建一个路径锁
		ZKPathLockData zkPLD = this.createLockPath(innerLockPath);
		String currentEphPath = zkPLD.getEphePath();

		if (this.isAlreadyGetLock(innerLockPath, currentEphPath)) {
			addScheduleDelPath(zkPLD,null,null);
			return zkPLD;
		} else {
			if(!this.waitGetLock(zkPLD, waitSeconds)){
				// 否则删除创建的临时节点，返回null
				this.release(zkPLD);
				return null;
			}
		}
		return zkPLD;
	}

	@Override
	public ZKPathLockData acquire(String lockPath) throws Exception {
		String innerLockPath = this.checkLockPath(lockPath);
		// 尝试创建一个路径锁
		ZKPathLockData zkPLD = this.createLockPath(innerLockPath);
		String currentEphPath = zkPLD.getEphePath();

		if (this.isAlreadyGetLock(innerLockPath, currentEphPath)) {
			addScheduleDelPath(zkPLD,null,null);
			return zkPLD;
		} else {
			if(!this.waitGetLock(zkPLD, null)){
				// 否则删除创建的临时节点，返回null
				this.release(zkPLD);
				return null;
			}
		}
		return zkPLD;
	}

	@Override
	public void release(ZKPathLockData lockData) throws Exception {
		
		String path=lockData.getLockPath();
		String secretCode=lockData.getLockData();
		String ephPath=lockData.getEphePath();
		
		if(!zkOperate.existsPath(ephPath)){
			throw new Exception("临时节点已不存在，"+ephPath);
		}
		byte[] data=zkOperate.getPathData(ephPath);
		if(null == data){
			throw new Exception("临时节点数据不存在，"+ephPath);
		}
		
		String pathDataStr=zkConnecter.getStrData(data);
		
		if(!secretCode.equals(pathDataStr)){
			throw new Exception("解锁数据与锁数据不一样，无法解锁");
		}else{
			zkOperate.deletNode(ephPath, -1);
			removeScheduleDelPath(lockData);
			logger.debug(path+" 释放锁,锁数据为 "+lockData.getLockData());
		}
		
	}

	/**
	 * 创建锁永久节点
	 * 
	 * @throws UnsupportedEncodingException
	 * @throws Exception
	 */
	private void createLockPersistentPath() throws UnsupportedEncodingException, Exception {
		zkOperate.createPath(LOCK_PERSISTENT_PATH, LOCK_PERSISTENT_PATH.getBytes("UTF-8"), true, CreateMode.PERSISTENT);
	}

	/**
	 * 某一个路径锁的完整路径
	 * 
	 * @param path
	 * @return
	 */
	private String getLockEphemeralPath(String path) {
		return LOCK_PERSISTENT_PATH + "/" + path;
	}

	/**
	 * 将一个路径生成内部的锁路径
	 * 
	 * @param path
	 * @return
	 */
	private String checkLockPath(String path) {
		if (null == path || path.trim().isEmpty()) {
			throw new NullPointerException();
		} else if (path.startsWith("/")) {
			path = path.replaceFirst("/*", "");
		}
		String lockPath = this.getLockEphemeralPath(path);
		return lockPath;
	}

	/**
	 * 创建一个锁路径的临时有序节点
	 * 
	 * @param path
	 * @return
	 * @throws Exception
	 */
	private ZKPathLockData createLockPath(String path) throws Exception {
		String joinPath = path + LOCK_SEPARATOR;
		String secretCode = this.createSecretCode();
		ZKPathLockData zkPLD = new ZKPathLockData();
		zkPLD.setLockData(secretCode);
		zkPLD.setLockPath(path);

		String ownEphemeralPath = zkOperate.createNode(joinPath, secretCode.getBytes("UTF-8"),
				CreateMode.EPHEMERAL_SEQUENTIAL);
		zkPLD.setEphePath(ownEphemeralPath);

		return zkPLD;
	}

	/**
	 * 获取比当前获取到的临时节点小一个的节点名
	 * 
	 * @param path
	 * @return
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	private String getLockPathSecondMinNumPath(String path, String ownEphemeralPath)
			throws KeeperException, InterruptedException {
		String joinPath = (path + LOCK_SEPARATOR).replaceFirst("/[^/]*/", "");
		String ownEphemeralSubPath = ownEphemeralPath.replaceFirst("/[^/]*/", "");
		String secondMinNumPath = null;

		List<String> childPathList = zkOperate.getChildPathList(LOCK_PERSISTENT_PATH);

		if (null == childPathList || childPathList.isEmpty()) {
			return null;
		}
		for (String childPath : childPathList) {
			if (!childPath.startsWith(joinPath)) {
				continue;
			}
			if (secondMinNumPath == null) {
				if (childPath.compareTo(ownEphemeralSubPath) < 0) {
					secondMinNumPath = childPath;
				}
			} else if (childPath.compareTo(ownEphemeralSubPath) < 0 && childPath.compareTo(secondMinNumPath) > 0) {
				secondMinNumPath = childPath;
			}
		}

		String resultPath = null == secondMinNumPath ? null
				: (ownEphemeralSubPath.equals(secondMinNumPath) ? null : secondMinNumPath);
		return null == resultPath ? null : (getLockEphemeralPath(resultPath));
	}

	/**
	 * 获取该路径下已经存在的路径
	 * 
	 * @param path
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	private List<String> getLockPathAlreadyLockPath(String path) throws KeeperException, InterruptedException {
		String joinPath = (path + LOCK_SEPARATOR).replaceFirst("/[^/]*/", "");

		List<String> childPathList = zkOperate.getChildPathList(LOCK_PERSISTENT_PATH);

		if (null == childPathList || childPathList.isEmpty()) {
			return null;
		}

		List<String> alereadyLockPathList = new LinkedList<String>();
		for (String childPath : childPathList) {
			if (!childPath.startsWith(joinPath)) {
				continue;
			}
			String fullChildPath = getLockEphemeralPath(childPath);
			alereadyLockPathList.add(fullChildPath);
		}

		return alereadyLockPathList;
	}

	/**
	 * 获取该路径创建的临时节点里最小的节点
	 * 
	 * @param path
	 * @return
	 * @throws InterruptedException 
	 * @throws KeeperException 
	 */
	private String getLockPathMinNumPath(String path) throws KeeperException, InterruptedException {
		String joinPath = (path + LOCK_SEPARATOR).replaceFirst("/[^/]*/", "");
		String minNumPath = null;
		List<String> childPathList = zkOperate.getChildPathList(LOCK_PERSISTENT_PATH);

		if (null == childPathList || childPathList.isEmpty()) {
			return null;
		}
		for (String childPath : childPathList) {
			if (!childPath.startsWith(joinPath)) {
				continue;
			}
			if (null == minNumPath) {
				minNumPath = childPath;
			} else if (childPath.compareTo(minNumPath) < 0) {
				minNumPath = childPath;
			}
		}

		return null == minNumPath ? null : (getLockEphemeralPath(minNumPath));
	}

	/**
	 * 创建一个独一无二的码 包含创建时间
	 * 
	 * @return
	 * @throws Exception
	 */
	private String createSecretCode() throws Exception {
		String objectStr = new Object().toString();
		String hostStr = zkConnecter.getCommander().getNodeManage().getCurrentHost().toString();

		String secretCode = hostStr + objectStr + "-" + System.currentTimeMillis();

		return secretCode;
	}

	/**
	 * 从创建的码里获取创建时间
	 * 
	 * @param secretCode
	 * @return
	 */
	private long fromSecretCodeGetStartTime(String secretCode) {
		String timeStr = secretCode.replaceFirst("^.*-", "");
		return Long.parseLong(timeStr);
	}

	/**
	 * 查看是否获取到了锁，当当前创建的有序节点是节点里存在的最小节点表示获取到了锁
	 * 
	 * @param currentLockPath
	 * @param currentEphPath
	 * @return
	 * @throws Exception
	 */
	private boolean isAlreadyGetLock(String currentLockPath, String currentEphPath) throws Exception {
		String minEphPath = this.getLockPathMinNumPath(currentLockPath);

		if (null == minEphPath) {
			throw new Exception("没有获取到最小临时有序节点，不正常状态");
		}

		return minEphPath.equals(currentEphPath);
	}

	/**
	 * @param zkPLD
	 * @param waitMilis
	 * @throws Exception
	 */
	private boolean waitGetLock(ZKPathLockData zkPLD, Long waitSeconds) throws Exception {

		String path = zkPLD.getLockPath();
		String ephPath = zkPLD.getEphePath();
		CountDownLatch latch = new CountDownLatch(1);
		try {
			addScheduleDelPath(zkPLD,latch,waitSeconds);
			
			if (isAlreadyGetLock(path,ephPath)) {
				latch.countDown();
			} else {
				getLock(zkPLD,latch);
			}
		} finally {
			latch.await();
		}

		return this.isAlreadyGetLock(path, ephPath);
	}

	
	/**获取锁
	 * @param zkPLD
	 * @param latch
	 * @param waitMilis
	 * @throws Exception 
	 */
	public void getLock(ZKPathLockData zkPLD,CountDownLatch latch) throws Exception{
		String path = zkPLD.getLockPath();
		String ephPath = zkPLD.getEphePath();
		
		String secondMinEphemeralPath = null;
		while(true){
			
			//如果信号已被消耗或创建的临时节点已经不存在则直接返回
			if(latch.await(1, TimeUnit.MILLISECONDS) || !zkOperate.existsPath(ephPath)){
				return;
			}
			//如果获取到了权限
			if(this.isAlreadyGetLock(path, ephPath)){
				latch.countDown();
				return;
			}
			
			while(true){
				secondMinEphemeralPath = this.getLockPathSecondMinNumPath(path, ephPath);
				// 如果没有获取到比自己小一个的节点则直接消耗
				if (null == secondMinEphemeralPath) {
					latch.countDown();
					return;
				}else if(!zkOperate.existsPath(secondMinEphemeralPath)){
					//如果获取到的节点已消失则重新获取
					continue;
				}else{
					break;
				}
			}
			
			Map<String,Object> callbackData=new HashMap<String,Object>(3);
			callbackData.put("zkLock", this);
			callbackData.put("latch", latch);
			callbackData.put("zkPLD", zkPLD);
			MyWatcher watcher = new MyWatcher(SingleInstanceComponent.getInstance(LockPathWatchCallback.class), callbackData,
					true);
			zkOperate.watchPath(secondMinEphemeralPath, watcher, false);
			if (!zkOperate.existsPath(secondMinEphemeralPath)) {
				//如果关注的节点已消失则重新获取
				continue;
			}else{
				break;
			}
		}
		
	}
	
	/**加入定时删除锁
	 * @param zkPLD
	 * @param latch
	 * @param waitSeconds
	 */
	private void addScheduleDelPath(ZKPathLockData zkPLD,CountDownLatch latch,Long waitSeconds){
		if(null == waitSeconds){
			waitSeconds=TimeUnit.HOURS.toSeconds(1);
		}
		//加入定时任务，如果出现异常则最长一小时后释放锁
		LockPathSchedule lps=new LockPathSchedule(this,zkPLD,latch,waitSeconds);
		lps.run();
	}
	
	/**移除定时任务
	 * @param zkPLD
	 */
	private void removeScheduleDelPath(ZKPathLockData zkPLD){
		LockPathSchedule lps=new LockPathSchedule(this,zkPLD,null,0);
		ScheduleUpdateUtil.removeJob(lps);
		
		logger.debug("定时删除锁任务被移除，"+lps.getId());
		lps.clear();
	}
}
